package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.service.IHotArticleService;
import com.heima.behavior.dto.UpdateArticleMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.security.Key;
import java.time.Duration;

/**
 * @author: itheima
 * @create: 2022-01-03 14:51
 */
@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {


    /**
     * 监听hot_article_score_topic话题中数据，数据流形式：{"add":1,"articleId":1474743178954362882,"type":1}-->{"add":1,"articleId":1474743178954362882,"type":2}->
     * 流式计算逻辑：统计最近一段时间（10秒以内）数据流中包含文章操作行为数据。
     *
     * @param input 数据流中数据：文章操作记录
     * @return 目标话题话题中数据形式-得到数据流中每篇文章操作结果：{"articleId"：1474743178954362882,view:5,like:3,comment:2,collect:1}
     */
    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        //1.处理数据--从数据库流（源头话题中）得到数据。重新设置数据Key（文章ID）   Val（文章信息）
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 重新改写 键值对 数据
             * @param key 数据中键
             * @param value 数据中值
             * @return 键值对对象
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1.1 获取原始数据流中JSON值，将json转为UpdateArticleMessage对象
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //1.2 重新设置数据key以及val
                return new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
            }
        });

        //2.对处理后数据：根据数据key(文章ID)进行分组
        KGroupedStream<String, String> groupedStream = map.groupByKey();

        //3.设置时间窗口：统计10秒内数据流中每篇文档所有的行为操作
        TimeWindowedKStream<String, String> windowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(20)));

        //4.设置聚合：10秒内数据流中所有的文章中，每篇文章多次行为操作进行聚合
        //4.1 设置初始消息信息 没有聚合结果消息
        Initializer<String> init = () -> {
            //第一条消息进入后，聚合结果为空
            return null;
        };
        //4.2 对10秒内每篇文章多次行为操作进行聚合
        Aggregator<String, String, String> agg = new Aggregator<String, String, String>() {
            /**
             * 聚合后回调函数，
             * @param key 数据流中数据Key 文章ID
             * @param value 数据流中数据 Val 用户行为
             * @param aggregate 聚合结果（上次聚合结果）
             * @return 当前文章聚合后结果（10秒内历史数据汇总）
             */
            @Override
            public String apply(String key, String value, String aggregate) {
                //4.2.1 声明文章聚合结果对象
                ArticleStreamMessage result = null;
                //4.2.2 判断上次聚合结果是否有值
                if (StringUtils.isBlank(aggregate)) {
                    //上次聚合结果为空，设置当前文章各种操作分数为0
                    result = new ArticleStreamMessage();
                    result.setArticleId(Long.parseLong(key));
                    result.setView(0);
                    result.setLike(0);
                    result.setComment(0);
                    result.setCollect(0);
                } else {
                    //说明上次聚合结果有值：结果{"articleId"：1474743178954362882,view:5,like:3,comment:2,collect:1}
                    result = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }

                //4.2.3 根据当前这次行为操作更新文章聚合结果对象
                UpdateArticleMessage updateArticleMessage = JSONObject.parseObject(value, UpdateArticleMessage.class);
                Integer add = updateArticleMessage.getAdd();
                switch (updateArticleMessage.getType()) {
                    case 0:
                        //阅读
                        result.setView(result.getView()+add);
                        break;
                    case 1:
                        // 点赞
                        result.setLike(result.getLike() + add);
                        break;
                    case 2:
                        // 评论
                        result.setComment(result.getComment() + add);
                        break;
                    case 3:
                        // 收藏
                        result.setCollect(result.getCollect() + add);
                        break;
                }
                //将聚合结果Result转为JSON
                return JSON.toJSONString(result);
            }
        };

        //4.x 设置聚合
        KTable<Windowed<String>, String> aggregate = windowedKStream.aggregate(init, agg);

        //5.将实时计算结果:数据流中所有文章，以及每篇文章所有行为操作，将操作结果发送目标话题中 {"articleId"：1474743178954362882,view:5,like:3,comment:2,collect:1}
        KStream<String, String> resultStream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return resultStream;
    }

}
